package com.ferry.transport.netty.provider;

import com.ferry.protocol.ProtocolDecoder;
import com.ferry.protocol.ProtocolEncoder;
import com.ferry.transport.api.config.ChildConfig;
import com.ferry.transport.api.config.ParentConfig;
import com.ferry.transport.netty.handler.server.AcceptorHandler;
import com.ferry.transport.netty.handler.server.AcceptorIdleStateTrigger;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollMode;
import io.netty.handler.timeout.IdleStateHandler;

import java.net.SocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Created by daiyong
 */
public class DefaultNettyTcpAcceptor extends NettyTcpAcceptor {

	private Map<String, String> tcpConfig = new HashMap<>();

	private ParentConfig parentConfig = new ParentConfig();

	private ChildConfig childConfig = new ChildConfig();

	private final ChannelOutboundHandler encoder = new ProtocolEncoder();

	private ProviderProcessor providerProcessor;

	public DefaultNettyTcpAcceptor(int port, int nWorkers) {
		super(port, nWorkers);
		this.tcpConfig = tcpConfig;
	}

	public DefaultNettyTcpAcceptor(SocketAddress localAddress, int nWorkers) {
		super(localAddress, nWorkers);
		this.tcpConfig = tcpConfig;
	}


	public DefaultNettyTcpAcceptor(int port, int nWorkers, boolean isNative) {
		super(port, nWorkers, isNative);
		this.tcpConfig = tcpConfig;
	}

	public DefaultNettyTcpAcceptor(SocketAddress localAddress, int nWorkers, boolean isNative) {
		super(localAddress, nWorkers, isNative);
		this.tcpConfig = tcpConfig;
	}

	@Override
	protected void init() {
		super.init();
	}

	@Override
	public ChannelFuture bind(SocketAddress localAddress) {
		ServerBootstrap boot = serverBootstrap();

		ParentConfig parentConfig = parentConfig();
		ChildConfig childConfig = childConfig();

		//设置boot 参数 参数太多 先验证功能
		boot.option(ChannelOption.SO_BACKLOG, parentConfig.getBacklog())
				.option(ChannelOption.SO_REUSEADDR, parentConfig.isReuseAddress())
				.option(EpollChannelOption.SO_REUSEPORT, parentConfig.isReusePort())
				.option(EpollChannelOption.IP_FREEBIND, parentConfig.isIpFreeBind())
				.option(EpollChannelOption.IP_TRANSPARENT, parentConfig.isIpTransparent());
		if (parentConfig.getRcvBuf() > 0) {
			boot.option(ChannelOption.SO_RCVBUF, parentConfig.getRcvBuf());
		}
		if (parentConfig.getPendingFastOpenRequestsThreshold() > 0) {
			boot.option(EpollChannelOption.TCP_FASTOPEN, parentConfig.getPendingFastOpenRequestsThreshold());
		}
		if (parentConfig.getTcpDeferAccept() > 0) {
			boot.option(EpollChannelOption.TCP_DEFER_ACCEPT, parentConfig.getTcpDeferAccept());
		}
		if (parentConfig.isEdgeTriggered()) {
			boot.option(EpollChannelOption.EPOLL_MODE, EpollMode.EDGE_TRIGGERED);
		} else {
			boot.option(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
		}

		// child options
		boot
				.childOption(ChannelOption.SO_REUSEADDR, childConfig.isReuseAddress())
				.childOption(ChannelOption.SO_KEEPALIVE, childConfig.isKeepAlive())
				.childOption(ChannelOption.TCP_NODELAY, childConfig.isTcpNoDelay())
				.childOption(ChannelOption.ALLOW_HALF_CLOSURE, childConfig.isAllowHalfClosure());
		if (childConfig.getRcvBuf() > 0) {
			boot.childOption(ChannelOption.SO_RCVBUF, childConfig.getRcvBuf());
		}
		if (childConfig.getSndBuf() > 0) {
			boot.childOption(ChannelOption.SO_SNDBUF, childConfig.getSndBuf());
		}
		if (childConfig.getLinger() > 0) {
			boot.childOption(ChannelOption.SO_LINGER, childConfig.getLinger());
		}
		if (childConfig.getIpTos() > 0) {
			boot.childOption(ChannelOption.IP_TOS, childConfig.getIpTos());
		}
		if (childConfig.getTcpNotSentLowAt() > 0) {
			boot.childOption(EpollChannelOption.TCP_NOTSENT_LOWAT, childConfig.getTcpNotSentLowAt());
		}
		if (childConfig.getTcpKeepCnt() > 0) {
			boot.childOption(EpollChannelOption.TCP_KEEPCNT, childConfig.getTcpKeepCnt());
		}
		if (childConfig.getTcpUserTimeout() > 0) {
			boot.childOption(EpollChannelOption.TCP_USER_TIMEOUT, childConfig.getTcpUserTimeout());
		}
		if (childConfig.getTcpKeepIdle() > 0) {
			boot.childOption(EpollChannelOption.TCP_KEEPIDLE, childConfig.getTcpKeepIdle());
		}
		if (childConfig.getTcpKeepInterval() > 0) {
			boot.childOption(EpollChannelOption.TCP_KEEPINTVL, childConfig.getTcpKeepInterval());
		}
		if (SocketChannelProvider.SocketType.NATIVE_EPOLL == socketType()) {
			boot.childOption(EpollChannelOption.TCP_CORK, childConfig.isTcpCork())
					.childOption(EpollChannelOption.TCP_QUICKACK, childConfig.isTcpQuickAck())
					.childOption(EpollChannelOption.IP_TRANSPARENT, childConfig.isIpTransparent());
			if (childConfig.isTcpFastOpenConnect()) {
				// Requires Linux kernel 4.11 or later
				boot.childOption(EpollChannelOption.TCP_FASTOPEN_CONNECT, childConfig.isTcpFastOpenConnect());
			}
			if (childConfig.isEdgeTriggered()) {
				boot.childOption(EpollChannelOption.EPOLL_MODE, EpollMode.EDGE_TRIGGERED);
			} else {
				boot.childOption(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
			}
		}

		initChannelFactory();

		boot.childHandler(new ChannelInitializer<Channel>() {
			@Override
			protected void initChannel(Channel ch) throws Exception {
				ch.pipeline().addLast(
						new IdleStateHandler(3, 0, 0, TimeUnit.SECONDS),
						new AcceptorIdleStateTrigger(),
						new ProtocolDecoder(),
						new ProtocolEncoder(),
						new AcceptorHandler(providerProcessor)
				);
			}
		});

		return boot.bind(localAddress);
	}

	@Override
	public void setProcessor(ProviderProcessor processor) {

	}

	@Override
	public void withProcessor(ProviderProcessor processor) {
		this.providerProcessor = processor;
	}

	@Override
	public ParentConfig parentConfig() {
		return parentConfig;
	}

	@Override
	public ChildConfig childConfig() {
		return childConfig;
	}
}
